package com.k8.common.executor;


import com.k8.common.aware.BeanFactoryAware;
import com.k8.common.executor.threadPool.ThreadPool;
import com.k8.common.factroy.BeanFactory;
import com.k8.rpc.RpcException;
import com.k8.rpc.url.URL;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

import static com.k8.common.constants.CommonKeyConstants.*;
import static com.k8.common.constants.SpiConstants.SMART_THREAD_POOL;
import static com.k8.common.executor.threadPool.AbstractThreadPool.SYSTEM_CORE;

public class DefaultExecutorRepository implements ExecutorRepository, BeanFactoryAware {
    BeanFactory beanFactory;
    private final ConcurrentHashMap<String, ConcurrentHashMap<String, ExecutorService>> data = new ConcurrentHashMap<>();
    //private final Lock createLock = new ReentrantLock(false);
    private static final String scheduled = "scheduledExecutorService";

    @Override
    public void beanFactoryAware(BeanFactory beanFactory) {
        this.beanFactory = beanFactory;
    }

    public ExecutorService getExecutorService(URL url) {
        String executorKey = getExecutorKey(url);
        ExecutorService executorService = getOrCreateExecutorService(executorKey, url);
        return executorService;
    }

    /*
     * 目前仅提供两种角色的不同的线程池
     * */
    private String getExecutorKey(URL url) {
        String side = url.getParameter(SIDE_KEY);
        if (side == null) return PROVIDER;
        if (!PROVIDER.equals(side) && !CONSUMER.equals(side)) throw new RpcException("Side error , side =" + side);
        return side;
    }

    private ExecutorService getOrCreateExecutorService(String executorKey, URL url) {
        ConcurrentHashMap<String, ExecutorService> executorsMap = data.get(executorKey);
        if (executorsMap == null) {
            data.computeIfAbsent(executorKey, key -> {
                return new ConcurrentHashMap<>();
            });
            executorsMap = data.get(executorKey);
        }
        String poolName = url.getParameter(THREAD_POOL_TYPE, SMART_THREAD_POOL);
        ExecutorService executorService = executorsMap.get(poolName);
        if (executorService == null) {
            synchronized (executorsMap) {
                if (executorsMap.get(poolName) == null) {
                    executorService = createExecutorService(url);
                    executorsMap.put(poolName, executorService);
                }
            }
        }
        return executorService;
    }

    private ExecutorService createExecutorService(URL url) {
        return beanFactory.getExtensionDirector().getAdaptiveExtension(ThreadPool.class).getExecutor(url);
    }

    @Override
    public ScheduledExecutorService getScheduledExecuteService(URL url) {
        int core = url.getParameter(CORE_THREAD_COUNT_KEY, SYSTEM_CORE * 2 + 1);
        String side = url.getSide();
        ConcurrentHashMap<String, ExecutorService> executorsMap = data.get(side);
        ExecutorService executorService = executorsMap.get(scheduled);
        if (executorService == null) {
            synchronized (executorsMap) {
                if (executorsMap.get(scheduled) == null) {
                    executorService = Executors.newScheduledThreadPool(core);
                    executorsMap.put(scheduled, executorService);
                }
            }
        }
        return (ScheduledExecutorService) executorService;
    }
}
